package com.ce.socket;

import java.lang.reflect.Constructor;
import java.util.Iterator;
import java.util.Vector;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

import org.apache.log4j.Logger;

import com.ce.study.socketpool.CM20MessageBase;

public class ABalance implements IBalance {
	private static Logger log = Logger.getLogger(ABalance.class);
	private SelfThread selfThread;
	private String[] servers;
	private Integer[] weights;
	private Vector<ServerWeight> sws;
	private boolean initialized = false;
	private ConcurrentMap<Integer, String> hosts;
	private ConcurrentMap<String, ISocketPool> pools;
	private Class<?> poolCls;
	private Class<?> socketCls;
	private int total = 0;

	public ABalance(String[] servers, Integer[] weights, Class<?> poolCls, Class<?> socketCls) {
		this.servers = servers;
		this.weights = weights;
		this.poolCls = poolCls;
		this.socketCls = socketCls;
	}
	
	public void initialize() {
		if(initialized) {
			log.error( "-> trying to initialize an already balancer" );
			return;
		}
		if(servers.length != weights.length) {
			log.error( "-> servers and weights size is not match" );
			return;
		}
			
		sws = new Vector<ServerWeight>();
		hosts = new ConcurrentHashMap<Integer, String>();
		pools = new ConcurrentHashMap<String, ISocketPool>();
		
		for (int i = 0; i < servers.length; i++) {
			ServerWeight sw = new ServerWeight(servers[i],weights[i]);
			sws.add(sw);
		}
		
		for (int i = 0; i < servers.length; i++) {
			Constructor[] cons = poolCls.getConstructors();
			Constructor matchcon = null;
			for (int j = 0; j < cons.length; j++) {
				if(cons[j].getParameterTypes().length == 2) {
					matchcon = cons[j];
					break;
				}
				
			}
			ISocketPool pool;
			try {
				pool = (ISocketPool) matchcon.newInstance(servers[i], socketCls);
			} catch (Exception e) {
				log.error("newInstance ISocketPool error", e);
				return;
			}
			//ISocketPool pool = new CSocketPool(servers[i],BSocket.class);
			pool.initialize();
			pools.put(servers[i], pool);
		}
		
		//构建
		constructHostMap();
		
		initialized = true;
		startSelfThread();
	}
	private boolean isInitialized() {
		return initialized;
	}
	
	private synchronized void constructHostMap() {
		int count = 0;
		for (Iterator<ServerWeight> iterator = sws.iterator(); iterator.hasNext();) {
			ServerWeight sw = iterator.next();
			ISocketPool pool = pools.get(sw.server);
			sw.isAvail = pool.isAvail();
			for (int i = 0; i < sw.weight; i++) {
				if(sw.isAvail) {
					hosts.put(count, sw.server);
					count += 1;
				}
			}
		}
		log.debug("total="+total+", count="+count);
		total = count;
//		total = 0;
//		hosts.clear();
//		for (Iterator<ServerWeight> iterator = sws.iterator(); iterator.hasNext();) {
//			ServerWeight sw = iterator.next();
//			ISocketPool pool = pools.get(sw.server);
//			sw.isAvail = pool.isAvail();
//			for (int i = 0; i < sw.weight; i++) {
//				if(sw.isAvail) {
//					hosts.put(total, sw.server);
//					total += 1;
//				}
//			}
//			
//		}
	}
	private ISocketPool getPool(int hashCode) throws DALException {
		if(total<=0) {
			checkPools();
			throw new DALException("has not avail pool");
		}
		Integer key = hashCode % total;
		log.debug("hashCode="+hashCode+", key="+key);
		String server = hosts.get(key);
		ISocketPool pool = pools.get(server);
		if(pool.isAvail())
			return pool;
		else {
			constructHostMap();
			if(total<=0) {
				throw new DALException("has not avail pool");
			} else {
				key = hashCode % total;
				server = hosts.get(key);
				pool = pools.get(server);
				return pool;
			}
		}
	}
	private void startSelfThread() {

		if ( selfThread != null ) {

			if ( selfThread.isRunning() ) {
				log.error( "=== self thread already running ===" );
			}
			else {
				selfThread.start();
			}
		}
		else {
			selfThread = new SelfThread( this );
			//selfThread.setInterval( this.daemonSleep );
			selfThread.start();
		}
	}
	private void checkPools() {
		for (Iterator<ISocketPool> iterator = pools.values().iterator(); iterator.hasNext();) {
			ISocketPool pool = iterator.next();
			pool.check();
		}
	}
	private void stopSelfThread() {
		if ( selfThread != null && selfThread.isRunning() )
			selfThread.stopThread();
	}
	public byte[] sendData(CM20MessageBase cm20) throws Exception {
		cm20.encode();
		ISocketPool socketPool = getPool(cm20.hashCode());
		return socketPool.sendData(cm20.getMessage(), 1);
	}
	public void selfDaemon() {
		constructHostMap();
	}
	private static class SelfThread extends Thread {

		// logger
		private static Logger log = Logger.getLogger( SelfThread.class );

		private ABalance balance;
		private long interval      = 1000 * 3; // every 3 seconds
		private boolean stopThread = false;
		private boolean running;

		protected SelfThread( ABalance balance ) {
			this.balance = balance;
			this.setDaemon( true );
			this.setName( "SelfThread" );
		}

		public void setInterval( long interval ) { this.interval = interval; }
		
		public boolean isRunning() {
			return this.running;
		}

		/** 
		 * sets stop variable
		 * and interupts any wait 
		 */
		public void stopThread() {
			this.stopThread = true;
			this.interrupt();
		}

		/** 
		 * Start the thread.
		 */
		public void run() {
			this.running = true;

			while ( !this.stopThread ) {
				try {
					// if pool is initialized, then
					// run the maintenance method on itself
					if ( balance.isInitialized() )
						balance.selfDaemon();
					
					Thread.sleep( interval );
				}
				catch ( Exception e ) {
					log.error("-> selfMaint error " + e);
				}
			}

			this.running = false;
		}
	}
	private class ServerWeight {
		String server;
		Integer weight;
		boolean isAvail = true;
		public ServerWeight(String server, Integer weight) {
			this.server = server;
			this.weight = weight;
		}
		public boolean isAvail() {
			return isAvail;
		}
		public void setAvail(boolean isAvail) {
			this.isAvail = isAvail;
		}
		public String getServer() {
			return server;
		}
		public Integer getWeight() {
			return weight;
		}
		
	}

}
